package com.googlecode.jlogalizer.mapred.log.map;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.log4j.Logger;

import com.googlecode.jlogalizer.mapred.log.Constants;
import com.googlecode.jlogalizer.mapred.util.LogParser;
import com.googlecode.jlogalizer.mapred.util.MapReducePropertyUtil;

public class LogMapper extends MapReduceBase implements Mapper<Object, Text, Text, IntWritable>  {

	private static Logger logger = Logger.getLogger(LogMapper.class);
	
	JobConf conf;
	@Override
	public void configure(JobConf job) {
		this.conf=job;
	}
	
	
	public void map(Object key, Text value, OutputCollector<Text, IntWritable> collector, Reporter reporter) throws IOException {

		String logEntryLine=value.toString();
		logger.debug("Log file type : "+conf.get("log_file_type"));
		
		String logEntryPattern=MapReducePropertyUtil.getInstance().getValue(conf.get("log_file_type"));
		logger.debug("Log entry pattern : "+logEntryPattern);
		
		int position=conf.getInt("position_in_log_file",0);

		String rawLog=LogParser.getInstance().getRawLogFromLogEntry(logEntryLine);
		String clientId=LogParser.getInstance().getClientIdFromLogEntry(logEntryLine);
		if (isValidLine(rawLog, logEntryPattern)){
			String ipAddress = retrieveIPAddress(rawLog, logEntryPattern, position);
			logger.debug("Ip address in map : "+ipAddress);
			collector.collect(new Text(clientId+"#" +ipAddress), Constants.ONE);
			//collector.collect(new Text(ipAddress), Constants.ONE);
			
		}
	}

	public final boolean isValidLine(String logEntryLine, String logEntryPattern){
		
		logger.debug("Log entry line : "+logEntryLine);
	    //String logEntryPattern = "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3}) (\\d+) \"([^\"]+)\" \"([^\"]+)\"";
	    Pattern p = Pattern.compile(logEntryPattern);
	    Matcher matcher = p.matcher(logEntryLine);
	    if (!matcher.matches()){
	    	//FIXME to use log4j
	    	System.err.print("Not a access log entry in File. Ignoring");
	    	return false;
		}
	    return true;
	}

	
	public final String retrieveIPAddress(String  logEntryLine, String logEntryPattern, int position){
		
	    //String logEntryPattern = "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3}) (\\d+) \"([^\"]+)\" \"([^\"]+)\"";
	    Pattern p = Pattern.compile(logEntryPattern);
	    Matcher matcher = p.matcher(logEntryLine);
	    if (!matcher.matches() || 
	      Constants.MIN_NUM_FIELDS != matcher.groupCount()) {
	      System.err.println("Invalid Log Entry");
	      System.err.println(logEntryLine);
	      return Constants.INVALID_IPADDRESS_STRING;
	    }
	    logger.debug("IP Address: " + matcher.group(position));
	    
	    return matcher.group(position);
	    //FIXME to be removed from this class soon after the rest of the map reduce tasks are done
	    /*logger.debug("Date&Time: " + matcher.group(4));
	    logger.debug("Request: " + matcher.group(5));
	    logger.debug("Response: " + matcher.group(6));
	    logger.debug("Bytes Sent: " + matcher.group(7));
	    if (!matcher.group(8).equals("-"))
	      logger.debug("Referer: " + matcher.group(8));
	    logger.debug("Browser: " + matcher.group(9));*/
		    
	}

}
